package com.winit.cloudlink.message;

import com.google.common.collect.Maps;
import com.winit.cloudlink.common.utils.Assert;
import com.winit.cloudlink.config.Metadata;

import java.util.Map;
import java.util.UUID;

import static com.winit.cloudlink.common.utils.StringUtils.isBlank;

/**
 * Created by stvli on 2015/11/12.
 */
public class MessageBuilder extends CloudlinkBuilder<Message> {

	private ExchangeType exchangeType = ExchangeType.Direct;
	private Object payload;
	private String fromApp;
	private String toApp;
	private String messageType;
	private String[] zones;
	private String routingKey;
	private String shardKey;

	public MessageBuilder(Metadata metadata) {
		super(metadata);
	}

	public MessageBuilder direct(String toApp, String messageType, Object payload) {
		return exchangeType(ExchangeType.Direct).fromApp(metadata.getApplicationOptions().getAppId2String())
				.toApp(toApp)
				.messageType(messageType)
				.payload(payload);
	}

	public MessageBuilder direct(String toApp, String messageType, String shardKey, Object payload) {
		return exchangeType(ExchangeType.Direct).fromApp(metadata.getApplicationOptions().getAppId2String())
				.toApp(toApp)
				.messageType(messageType)
				.shardKey(shardKey)
				.payload(payload);
	}

	public MessageBuilder topic(String messageType, Object payload, String routingKey, String... zones) {
		return exchangeType(ExchangeType.Topic).fromApp(metadata.getApplicationOptions().getAppId2String())
				.messageType(messageType)
				.routingKey(routingKey)
				.zones(zones)
				.payload(payload);
	}


	public MessageBuilder toApp(String toApp) {
		Assert.notNull(toApp, "'toApp' cannot be blank.");
		this.toApp = toApp;
		return this;
	}

	public MessageBuilder messageType(String messageType, String shardKey) {
		Assert.notBlank(messageType, "'messageType' cannot be blank.");
		Assert.notBlank(shardKey, "'shardKey' cannot be blank.");
		this.messageType = messageType;
		this.shardKey = shardKey;
		return this;
	}

	public MessageBuilder messageType(String messageType) {
		Assert.notBlank(messageType, "'messageType' cannot be blank.");
		this.messageType = messageType;
		return this;
	}

	public MessageBuilder shardKey(String shardKey) {
		Assert.notBlank(shardKey, "'shardKey' cannot be blank.");
		this.shardKey = shardKey;
		return this;
	}

	public MessageBuilder payload(Object payload) {
		Assert.notNull(payload, "'payload' cannot be blank.");
		this.payload = payload;
		return this;
	}

	public MessageBuilder zones(String[] zones) {
		this.zones = zones;
		return this;
	}

	protected MessageBuilder exchangeType(ExchangeType exchangeType) {
		Assert.notNull(exchangeType, "'exchangeType' cannot be blank.");
		this.exchangeType = exchangeType;
		return this;
	}

	protected MessageBuilder fromApp(String fromApp) {
		Assert.notNull(fromApp, "'fromApp' cannot be blank.");
		this.fromApp = fromApp;
		return this;
	}

	protected MessageBuilder routingKey(String routingKey) {
		Assert.notNull(toApp, "'toApp' cannot be blank.");
		this.routingKey = routingKey;
		return this;
	}

	@Override
	public Message build() {
		if (exchangeType == null) {
			throw new IllegalStateException("The 'exchangeType' property of message must not be null.");
		}
		if (isBlank(messageType)) {
			throw new IllegalStateException("The 'messageType' property of message must not be blank.");
		}
		if (ExchangeType.Direct.equals(exchangeType) && isBlank(toApp)) {
			throw new IllegalStateException("The 'toApp' property of direct message must not be blank.");
		}

		Map<String, Object> headers = Maps.newHashMap();
		headers.put(MessageHeaders.KEY_EXCHANGE_TYPE, exchangeType);
		headers.put(MessageHeaders.KEY_TO_APP, toApp);

		if (isBlank(fromApp)) {
			this.fromApp = metadata.getApplicationOptions().getAppId2String();
		}
		headers.put(MessageHeaders.KEY_FROM_APP, fromApp);

		if (ExchangeType.Direct.equals(exchangeType) && !isBlank(shardKey)) {
			this.messageType = this.messageType + Constants.MESSAGE_SHARD_KEY_SEPARATOR + shardKey;
		}
		headers.put(MessageHeaders.KEY_MESSAGETYPE, messageType);

		headers.put(MessageHeaders.KEY_SHARD_KEY, shardKey);
		headers.put(MessageHeaders.KEY_MESSAGE_ID, UUID.randomUUID().toString());
		if (zones != null && zones.length > 0) {
			headers.put(MessageHeaders.KEY_ZONES, zones);
		}

		if (isBlank(this.routingKey)) {
			this.routingKey = buildRoutingKey(toApp, messageType, zones);
		}
		headers.put(MessageHeaders.KEY_ROUTING_KEY, routingKey);
		return new Message(payload, new MessageHeaders(headers));
	}

	private String buildRoutingKey(String toAppId, String messageType, String[] zones) {
		StringBuffer sb = new StringBuffer();
		if (ExchangeType.Direct.equals(exchangeType)) {
			sb.append(messageType).append(Constants.MESSAGE_ROUTING_KEY_SEPARATOR).append(toAppId);
		} else {
			sb.append(messageType)
					.append(Constants.MESSAGE_ROUTING_KEY_SEPARATOR)
					.append(Constants.EVENT_QUEUE_KEYWORD);
			if (zones != null && zones.length > 0) {
				for (String zone : zones) {
					if (!isBlank(zone)) {
						sb.append(Constants.MESSAGE_ROUTING_KEY_SEPARATOR).append(zone);
					}
				}
			}
		}
		return sb.toString();
	}
}
